package com.streamsets.datacollector.pileline.stage;

import com.streamsets.datacollector.bundles.Constants;
import com.streamsets.datacollector.pileline.enums.HiveQueryConfigEnum;
import com.streamsets.datacollector.pileline.enums.ImpalaQueryConfigEnum;
import com.streamsets.datacollector.restapi.bean.ConfigConfigurationJson;
import com.streamsets.datacollector.restapi.bean.StageConfigurationJson;
import org.apache.commons.lang.StringUtils;

import java.util.ArrayList;
import java.util.List;

/**
 * 初始化impala查询配置
 * @author wengzhonghui
 * @date 2019/10/16 19:12
 */
public class InitTargetOfHiveMergeConfig {

    /*
     * hive query stage名称
     */
    public static final String STAGE_HIVE_QUERY_NAME = "HiveQuery_02";

    public static void init(StageConfigurationJson stage, String defaultImpalaJdbcUrl
            , String targetTbName, String updateColumn, boolean isIncMode, String priColumn, String defaultHiveUsername, String defaultHivePassword){
        if(targetTbName!=null){
            targetTbName = targetTbName.toLowerCase();
        }
        if(STAGE_HIVE_QUERY_NAME.equalsIgnoreCase(stage.getInstanceName())) {
            if (targetTbName != null) {
                List<ConfigConfigurationJson> configs =
                    intConfig(stage.getConfiguration(), defaultImpalaJdbcUrl, targetTbName, updateColumn, isIncMode,
                        priColumn, defaultHiveUsername, defaultHivePassword);
                stage.setConfiguration(configs);
            }
        }
    }

    public static List<ConfigConfigurationJson> intConfig(List<ConfigConfigurationJson> oldConfigs
            , String defaultImpalaJdbcUrl, String targetTbName, String updateColumn, boolean isIncMode, String priColumn, String defaultHiveUsername, String defaultHivePassword){
        List<ConfigConfigurationJson> newConfigs = new ArrayList<>();
        for(ConfigConfigurationJson config : oldConfigs ){
            config = initImpalaJDBCUrl(config, defaultImpalaJdbcUrl);
            config = initImpalaJDBCUsername(config, defaultHiveUsername);
            config = initImpalaJDBCPassword(config, defaultHivePassword);
            config = initImpalaDriver(config);
            config = initImpalaQuery(config, targetTbName, updateColumn, isIncMode,priColumn);
            config = initTargetTbName(config,targetTbName);
            newConfigs.add(config);
        }
        return newConfigs;
    }


    // 初始化hiveJdbc连接
    private static ConfigConfigurationJson initTargetTbName(ConfigConfigurationJson config, String targetTbName){
        if(ImpalaQueryConfigEnum.targetTable.getName().equalsIgnoreCase(config.getName())){
            config = new ConfigConfigurationJson(ImpalaQueryConfigEnum.targetTable.getName()
                    ,targetTbName);
        }
        return config;
    }

    // 初始化hiveJdbc连接
    private static ConfigConfigurationJson initImpalaJDBCUrl(ConfigConfigurationJson config, String defaultHiveJdbcUrl){
        if(ImpalaQueryConfigEnum.impalaJDBCUrl.getName().equalsIgnoreCase(config.getName())){
            config = new ConfigConfigurationJson(ImpalaQueryConfigEnum.impalaJDBCUrl.getName(),defaultHiveJdbcUrl);
        }
        return config;
    }
    private static ConfigConfigurationJson initImpalaJDBCUsername(ConfigConfigurationJson config, String defaultHiveUsername){
        if(ImpalaQueryConfigEnum.impalaJDBCUsername.getName().equalsIgnoreCase(config.getName())){
            config = new ConfigConfigurationJson(ImpalaQueryConfigEnum.impalaJDBCUsername.getName(),defaultHiveUsername);
        }
        return config;
    }
    private static ConfigConfigurationJson initImpalaJDBCPassword(ConfigConfigurationJson config, String defaultHivePassword){
        if(ImpalaQueryConfigEnum.impalaJDBCPassword.getName().equalsIgnoreCase(config.getName())){
            config = new ConfigConfigurationJson(ImpalaQueryConfigEnum.impalaJDBCPassword.getName(),defaultHivePassword);
        }
        return config;
    }

    // 初始化hive驱动
    private static ConfigConfigurationJson initImpalaDriver(ConfigConfigurationJson config){
        if(ImpalaQueryConfigEnum.hiveJDBCDriver.getName().equalsIgnoreCase(config.getName())){
            config = new ConfigConfigurationJson(HiveQueryConfigEnum.hiveJDBCDriver.getName()
                    ,HiveQueryConfigEnum.hiveJDBCDriver.getDefaultValue());
        }
        return config;
    }

    // 初始化增量合并的功能
    private static ConfigConfigurationJson initImpalaQuery(ConfigConfigurationJson config, String targetTbName
            , String updateColumn, boolean isIncMode, String priColumn){
        if(ImpalaQueryConfigEnum.queries.getName().equalsIgnoreCase(config.getName())){
            // 增量模式
            List<String> querys = null;
            if(isIncMode){
                querys = getQuerysForIncMode(targetTbName, updateColumn, priColumn);
            }else{
                // 全量模式
                querys = getQuerysForFullMode(targetTbName);
            }
            config = new ConfigConfigurationJson(ImpalaQueryConfigEnum.queries.getName()
                    ,querys);
        }
        return config;
    }


    /*
     * 获取增量合并的SQL
     * @param [targetTbName, updateColumn]
     * @return
     */
    private  static List<String> getQuerysForFullMode(String targetTbName){
        String tableInc = Constants.PROPERTIES_OLLECT_TEMP_TABLE_PREFIX  + targetTbName + "_inc";
        String tableIncTemp = Constants.PROPERTIES_OLLECT_TEMP_TABLE_PREFIX  + targetTbName + "_inc_temp";
        List<String> querys = new ArrayList<>();
//        querys.add("INVALIDATE METADATA " + tableInc);
//        querys.add("create table  if not exists "+ tableInc +" like "+ tableIncTemp +" STORED AS AVRO ");
        querys.add("create table  if not exists "+ targetTbName +" like "+ tableInc +" STORED AS AVRO ");
        querys.add("DROP TABLE IF EXISTS " + tableIncTemp);
        querys.add("ALTER TABLE "+ tableInc +" RENAME TO " + tableIncTemp);
        querys.add("CREATE TABLE "+ tableInc +" LIKE "+ targetTbName +" STORED AS AVRO  ");
        querys.add("drop table IF EXISTS " + targetTbName);
        querys.add("ALTER TABLE "+ tableIncTemp +" RENAME TO "+ targetTbName);
//        querys.add("INVALIDATE METADATA " + targetTbName);
//        querys.add("INVALIDATE METADATA " + tableInc);
        querys.add("DROP TABLE IF EXISTS " + tableIncTemp);
        return querys;
    }

    /*
     * 获取增量合并的SQL
     * @param [targetTbName, updateColumn]
     * @return
     */
    private  static List<String> getQuerysForIncMode(String targetTbName, String updateColumn, String priColumn){
        String tableInc = Constants.PROPERTIES_OLLECT_TEMP_TABLE_PREFIX  + targetTbName + "_inc";
        String tableIncTemp = Constants.PROPERTIES_OLLECT_TEMP_TABLE_PREFIX  + targetTbName + "_inc_temp";
        String tableTemp = Constants.PROPERTIES_OLLECT_TEMP_TABLE_PREFIX  + targetTbName + "_temp";
        List<String> querys = new ArrayList<>();
//        querys.add("INVALIDATE METADATA " + tableInc);
//        querys.add("create table  if not exists "+ tableInc +" like "+ tableIncTemp +" STORED AS AVRO ");
        querys.add("create table  if not exists "+ targetTbName +" like "+ tableInc +" STORED AS AVRO ");
        querys.add("DROP TABLE IF EXISTS " + tableIncTemp);
        querys.add("ALTER TABLE "+ tableInc +" RENAME TO " + tableIncTemp);
        querys.add("CREATE TABLE "+ tableInc +" LIKE "+ targetTbName +" STORED AS AVRO  ");
        querys.add("drop table IF EXISTS " + tableTemp);
        querys.add(getSqlForIncMerge(targetTbName, updateColumn, priColumn));
        querys.add("drop table IF EXISTS " + targetTbName);
        querys.add("ALTER TABLE "+ tableTemp +" RENAME TO "+ targetTbName);
//        querys.add("INVALIDATE METADATA " + targetTbName);
//        querys.add("INVALIDATE METADATA " + tableInc);
        querys.add("DROP TABLE IF EXISTS " + tableIncTemp);
        return querys;
    }



    /*
     * 获取增量合并的SQL
     * @param [targetTbName, updateColumn]
     * @return
     */
    private static String getSqlForIncMerge(String targetTbName, String updateColumn, String priColumn){
        priColumn = StringUtils.isEmpty(priColumn) ? "id" : priColumn;
        String tableIncTemp = Constants.PROPERTIES_OLLECT_TEMP_TABLE_PREFIX  + targetTbName + "_inc_temp";
        String tableTemp = Constants.PROPERTIES_OLLECT_TEMP_TABLE_PREFIX  + targetTbName + "_temp";
        //构建增量合并SQL，完整语句如：create table base_temp as SELECT t1.* FROM (SELECT * FROM base_full UNION ALL SELECT * FROM base_inc_temp) t1 JOIN(SELECT id, max(modifieddate) max_modified FROM(SELECT * FROM base_full UNION ALL SELECT * FROM base_inc_temp) t2 GROUP BY id) s ON t1.id = s.id AND t1.modifieddate = s.max_modified
        StringBuilder mergeSql = new StringBuilder();
        mergeSql.append("create table ");
        mergeSql.append(tableTemp);
        mergeSql.append(" as SELECT t1.* FROM (SELECT * FROM  ");
        mergeSql.append(targetTbName);
        mergeSql.append("   UNION  SELECT * FROM  ");
        mergeSql.append(tableIncTemp);
        mergeSql.append(" ) t1 JOIN(SELECT "+priColumn+", max(");
        mergeSql.append(updateColumn);
        mergeSql.append(") max_modified FROM(SELECT * FROM ");
        mergeSql.append(targetTbName);
        mergeSql.append(" UNION ALL SELECT * FROM ");
        mergeSql.append(tableIncTemp);
        mergeSql.append(") t2 GROUP BY "+priColumn+") s ON t1."+priColumn+" = s."+priColumn+" AND t1.");
        mergeSql.append(updateColumn);
        mergeSql.append(" = s.max_modified");
        return mergeSql.toString();
    }
}
